package com.sqoop.core.tools;

import net.sf.json.JSONObject;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.io.Serializable;

public class MsgQueueSenderUtils {
	
	private static Logger logger = LoggerFactory.getLogger(MsgQueueSenderUtils.class.getName());
	
	private  static final String username = ActiveMQConnection.DEFAULT_USER;
	private  static final String passwd = ActiveMQConnection.DEFAULT_PASSWORD;
	
	private static final String queuetype = "0";
	private static final String topictype = "1";
	
	/**
	 * 发送数据到队列
	 * @param url 队列url
	 * @param queuename 队列名称
	 * @param data 数据 
	 * @return 成功true 失败false
	 */
	public static boolean sendMsg(String url,String queuename,String queuetype,Object data){
		return sendMsg(url, username, passwd, queuename,queuetype, data);
		
	}
	
	/**
	 * 发送数据到队列 
	 * @param url url 队列url
	 * @param username 用户名
	 * @param passwd 密码
	 * @param queuename 队列名称
	 * @param data 数据
	 * @return 成功true 失败false
	 */
	public static boolean sendMsg(String url,String username,String passwd,String queuename,String queuetype,Object data){
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(username, passwd, url);
		connectionFactory.setTrustAllPackages(true);
		Destination destination = null;
		Session session = null;
		MessageProducer producer = null;
		Connection connection  = null;
		try{
			connection = connectionFactory.createConnection(); 
			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			
			if(queuetype.equals(topictype)){
				destination = session.createTopic(queuename);
			}else{
				destination = session.createQueue(queuename);
			}			
			producer = session.createProducer(destination);
			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
			//session.
			sendMsg(session,producer,data);
			session.commit();
		}catch(Exception e){
			logger.error("send msg error with e" + e);
			e.printStackTrace();
			return false;
		}finally{
			try {
				if(connection != null){
					connection.close();
				}
			} catch (JMSException e) {
				e.printStackTrace();
			}
			try {
				session.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		
		return true;
		
	}
	
	/**
	 * 发送数据
	 * @param session
	 * @param producer
	 * @param data
	 * @throws JMSException
	 */
	private static void sendMsg(Session session, MessageProducer producer,Object data) throws JMSException {
		if(data instanceof JSONObject){
			JSONObject jsonObject = (JSONObject)data;
			TextMessage texmessage = session.createTextMessage(jsonObject.toString());
			producer.send(texmessage);
		}
		else if(data instanceof CharSequence){
			TextMessage texmessage = session.createTextMessage(String.valueOf(data));
			producer.send(texmessage);
		}else if(data instanceof Serializable){
			ObjectMessage  msg = session.createObjectMessage();
			ObjectWrapper wapper = new ObjectWrapper();
			wapper.setObj(data);
			msg.setObject(wapper);
			producer.send(msg);;
		}else{
			logger.info("data type not parse " + data.getClass().getName());
		}
	}
	
	/**
	 * 包裹类
	 * @author zw
	 *
	 */
	private  static class ObjectWrapper implements Serializable{
		public Object obj;
		public void setObj(Object obj) {
			this.obj = obj;
		}
		public Object getObj() {
			return obj;
		}
	}
	
}
